﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace RMQLab.Worker
{
    class Program
    {
        static void Main(string[] args)
        {
            //Receive();
            //Subscribe();
            Task.Factory.StartNew(Compensate);
            Console.ReadLine();
            SubWithRoute(args);
        }
        static void Receive()
        {
            int threadId = Thread.CurrentThread.ManagedThreadId;
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("hello", false, false, false, null);
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume("hello", true, consumer);

                    Console.WriteLine(" [*] Waiting for messages. " +
                                      "To exit press CTRL+C");
                    while (true)
                    {
                        var ea =
                            (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine(" [x] Received {0}", message);

                        int dots = message.Split('.').Length - 1;
                        Thread.Sleep(dots * 1000);

                        Console.WriteLine(" [x] Done");
                    }

                }
            }
        }

        static void Subscribe()
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    string queueName = channel.QueueDeclare().QueueName;
                    channel.ExchangeDeclare("logs", "fanout");
                    channel.QueueBind(queueName, "logs", string.Empty);
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume(queueName, true, consumer);

                    Console.WriteLine(" [*] Waiting for messages. " +
                                      "To exit press CTRL+C");
                    while (true)
                    {
                        BasicDeliverEventArgs ea = consumer.Queue.Dequeue();

                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);

                        Console.WriteLine(" [x] {0}", message);
                    }

                }
            }
        }

        static void SubWithRoute(string[] args)
        {
            if (args.Length == 0)
            {
                args = new string[] { "info", "warn", "critical" };
                //Console.WriteLine("input severities with space splitting");
                //string argStr = Console.ReadLine();
                //args = argStr.Split(' ');
                //if (args.Length==0)
                //{
                //    Console.WriteLine("bye");
                //    return;
                //}
            }
            var workQueueArgs = new Dictionary<string, object> {
            { "x-dead-letter-exchange", "retry-logs-routed" },
            {"x-message-ttl",10}
            };
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    string queueName = "logs-queue";
                    channel.QueueDelete(queueName,false,false);
                    channel.QueueDeclare(queueName,false,false,false,workQueueArgs);
                    channel.ExchangeDeclare("logs-routed", "direct");
                    for (int i = 0; i < args.Length; i++)
                    {
                        channel.QueueBind(queueName, "logs-routed", args[i]);
                        Console.WriteLine("bound to routing key {0}", args[i]);
                    }
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume(queueName, false, consumer);

                    Console.WriteLine(" [*] Waiting for messages. " +
                                      "To exit press CTRL+C");
                    int j = 0;
                    while (true)
                    {
                        //Thread.Sleep(200);
                        BasicDeliverEventArgs ea = consumer.Queue.Dequeue();
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                        int index = message.IndexOf(":");
                        string msg = message.Substring(0, index);
                        double length = 2 * message.Length / (1024 * 1024);
                        Console.WriteLine(" [x] {0}: {1} Mb, redelivered: {2}", msg, length, ea.Redelivered);
                        try
                        {
                            if (j % 2 != 0)
                            {
                                channel.BasicAck(ea.DeliveryTag, false);
                            }
                            else
                            {
                                throw new Exception("fake ex");
                            }
                        }
                        catch (Exception)
                        {
                            Console.WriteLine(" [x] Nack {0}: {1} Mb", msg, length);
                            channel.BasicNack(ea.DeliveryTag, false, false);
                        }
                        j++;
                    }

                }
            }
        }

        static void Compensate()
        {
            var queueArgs = new Dictionary<string, object> 
            {               
                {"x-dead-letter-exchange", "logs-routed" },
                { "x-message-ttl", 20 }
            };
            var factory = new ConnectionFactory { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    string queueName = "retry-logs-queue";
                    string exchangeName ="retry-logs-routed";
                    channel.ExchangeDelete(exchangeName);
                    channel.ExchangeDeclare(exchangeName, "fanout");
                    channel.QueueDelete(queueName);
                    channel.QueueDeclare(queueName, true, false, false, queueArgs);
                    channel.QueueBind("retry-logs-queue", "retry-logs-routed", string.Empty);

                    //channel.QueueBind("retry-logs-queue", "retry-logs-routed", "info");
                    //channel.QueueBind("retry-logs-queue", "retry-logs-routed", "warn");
                    //channel.QueueBind("retry-logs-queue", "retry-logs-routed", "critical");
                            var consumer = new QueueingBasicConsumer(channel);
                            channel.BasicConsume(queueName, true, consumer);
                    Console.WriteLine(" [*] Waiting for dead letter messages. ");
                    int i = 0;
                    while (true)
                    {
                        //if (i % 2 != 0)
                        //{
                            //Thread.Sleep(3000);
                            BasicDeliverEventArgs ea = consumer.Queue.Dequeue();
                            string message = Encoding.UTF8.GetString(ea.Body);
                            int index = message.IndexOf(":");
                            string msg = message.Substring(0, index);
                            double length = 2 * message.Length / (1024 * 1024);
                            Console.WriteLine(" [x] dead letter queue: {0} {1} Mb", msg, length);
                        //}
                        //else
                        //{
                        //    Thread.Sleep(300);

                        //}
                        i++;
                    }
                }
            }
        }
    }
}
